package com.atguigu.edu.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.DimJoinFunction;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import com.atguigu.edu.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T> {
    private Connection hbaseConn;
    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HBaseUtil.getHbaseConnection();
        jedis = RedisUtil.getJedis();
    }
    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hbaseConn);
        RedisUtil.closeJedis(jedis);
    }
    @Override
    public T map(T obj) throws Exception {
        //根据流中对象获取要关联的维度主键
        String rowKey =getRowKey(obj);
        //根据流中的主键去redis中获取对应的维度数据
        JSONObject dimJsonObj  = RedisUtil.readDim(jedis, getTableName(), rowKey);
        //如果redis中不存在维度数据，去hbase中找
        if(dimJsonObj!=null){
            System.out.println("~~~从Redis中获取到了"+getTableName()+"表的"+rowKey+"维度数据~~~");
        }else {
            dimJsonObj = HBaseUtil.readRow(hbaseConn, Constant.HBASE_NAMESPACE, getTableName(), rowKey, JSONObject.class);
            if (dimJsonObj!=null){
                System.out.println("~~~从HBase中获取到了"+getTableName()+"表的"+rowKey+"维度数据~~~");
                RedisUtil.writeDim(jedis,getTableName(),rowKey,dimJsonObj);
            }else {
                System.out.println("~~~要查询的"+getTableName()+"表的"+rowKey+"维度数据不存在~~~");
            }
        }
        //将维度属性补充到流中对象上
        if (dimJsonObj!=null){
            addDims(obj,dimJsonObj);
        }


        return obj;
    }

}
